
<!doctype html>
<html lang="en" class="no-js">
  <head>
    
      <meta charset="utf-8">
      <meta name="viewport" content="width=device-width,initial-scale=1">
      
      
      
      <link rel="icon" href="../../../../static/images/favicon.png">
      <meta name="generator" content="mkdocs-1.3.0, mkdocs-material-8.2.8">
    
    
      
        <title>生产环境 kafka 消费中断如何 offset 重放消费 - WL4G DOCS</title>
      
    
    
      <link rel="stylesheet" href="../../../../assets/stylesheets/main.644de097.min.css">
      
        
        <link rel="stylesheet" href="../../../../assets/stylesheets/palette.e6a45f82.min.css">
        
      
    
    
    
      
        
        
        <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
        <link rel="stylesheet" href="https://fonts.googleapis.com/css?family=Roboto:300,300i,400,400i,700,700i%7CRoboto+Mono:400,400i,700,700i&display=fallback">
        <style>:root{--md-text-font:"Roboto";--md-code-font:"Roboto Mono"}</style>
      
    
    
      <link rel="stylesheet" href="../../../../static/css/util.css">
    
    <script>__md_scope=new URL("../../../..",location),__md_get=(e,_=localStorage,t=__md_scope)=>JSON.parse(_.getItem(t.pathname+"."+e)),__md_set=(e,_,t=localStorage,a=__md_scope)=>{try{t.setItem(a.pathname+"."+e,JSON.stringify(_))}catch(e){}}</script>
    
      

    
    
  </head>
  
  
    
    
      
    
    
    
    
    <body dir="ltr" data-md-color-scheme="default" data-md-color-primary="" data-md-color-accent="">
  
    
    
      <script>var palette=__md_get("__palette");if(palette&&"object"==typeof palette.color)for(var key of Object.keys(palette.color))document.body.setAttribute("data-md-color-"+key,palette.color[key])</script>
    
    <input class="md-toggle" data-md-toggle="drawer" type="checkbox" id="__drawer" autocomplete="off">
    <input class="md-toggle" data-md-toggle="search" type="checkbox" id="__search" autocomplete="off">
    <label class="md-overlay" for="__drawer"></label>
    <div data-md-component="skip">
      
        
        <a href="#kafka-offset" class="md-skip">
          Skip to content
        </a>
      
    </div>
    <div data-md-component="announce">
      
    </div>
    
      <div data-md-component="outdated" hidden>
        <aside class="md-banner md-banner--warning">
          
        </aside>
      </div>
    
    
      

<header class="md-header" data-md-component="header">
  <nav class="md-header__inner md-grid" aria-label="Header">
    <a href="../../../.." title="WL4G DOCS" class="md-header__button md-logo" aria-label="WL4G DOCS" data-md-component="logo">
      
  <img src="../../../../static/images/mylogo.jpeg" alt="logo">

    </a>
    <label class="md-header__button md-icon" for="__drawer">
      <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M3 6h18v2H3V6m0 5h18v2H3v-2m0 5h18v2H3v-2z"/></svg>
    </label>
    <div class="md-header__title" data-md-component="header-title">
      <div class="md-header__ellipsis">
        <div class="md-header__topic">
          <span class="md-ellipsis">
            WL4G DOCS
          </span>
        </div>
        <div class="md-header__topic" data-md-component="header-topic">
          <span class="md-ellipsis">
            
              生产环境 kafka 消费中断如何 offset 重放消费
            
          </span>
        </div>
      </div>
    </div>
    
      <form class="md-header__option" data-md-component="palette">
        
          
          
          <input class="md-option" data-md-color-media="(prefers-color-scheme: light)" data-md-color-scheme="default" data-md-color-primary="" data-md-color-accent=""  aria-label="Switch to dark mode"  type="radio" name="__palette" id="__palette_1">
          
            <label class="md-header__button md-icon" title="Switch to dark mode" for="__palette_2" hidden>
              <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M17 6H7c-3.31 0-6 2.69-6 6s2.69 6 6 6h10c3.31 0 6-2.69 6-6s-2.69-6-6-6zm0 10H7c-2.21 0-4-1.79-4-4s1.79-4 4-4h10c2.21 0 4 1.79 4 4s-1.79 4-4 4zM7 9c-1.66 0-3 1.34-3 3s1.34 3 3 3 3-1.34 3-3-1.34-3-3-3z"/></svg>
            </label>
          
        
          
          
          <input class="md-option" data-md-color-media="(prefers-color-scheme: dark)" data-md-color-scheme="slate" data-md-color-primary="" data-md-color-accent=""  aria-label="Switch to light mode"  type="radio" name="__palette" id="__palette_2">
          
            <label class="md-header__button md-icon" title="Switch to light mode" for="__palette_1" hidden>
              <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M17 7H7a5 5 0 0 0-5 5 5 5 0 0 0 5 5h10a5 5 0 0 0 5-5 5 5 0 0 0-5-5m0 8a3 3 0 0 1-3-3 3 3 0 0 1 3-3 3 3 0 0 1 3 3 3 3 0 0 1-3 3z"/></svg>
            </label>
          
        
      </form>
    
    
      <div class="md-header__option">
        <div class="md-select">
          
          <button class="md-header__button md-icon" aria-label="Select language">
            <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="m12.87 15.07-2.54-2.51.03-.03A17.52 17.52 0 0 0 14.07 6H17V4h-7V2H8v2H1v2h11.17C11.5 7.92 10.44 9.75 9 11.35 8.07 10.32 7.3 9.19 6.69 8h-2c.73 1.63 1.73 3.17 2.98 4.56l-5.09 5.02L4 19l5-5 3.11 3.11.76-2.04M18.5 10h-2L12 22h2l1.12-3h4.75L21 22h2l-4.5-12m-2.62 7 1.62-4.33L19.12 17h-3.24z"/></svg>
          </button>
          <div class="md-select__inner">
            <ul class="md-select__list">
              
                <li class="md-select__item">
                  <a href="/en/" hreflang="en" class="md-select__link">
                    English
                  </a>
                </li>
                
                <li class="md-select__item">
                  <a href="/zh/" hreflang="zh" class="md-select__link">
                    简体中文
                  </a>
                </li>
                
            </ul>
          </div>
        </div>
      </div>
    
    
      <label class="md-header__button md-icon" for="__search">
        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M9.5 3A6.5 6.5 0 0 1 16 9.5c0 1.61-.59 3.09-1.56 4.23l.27.27h.79l5 5-1.5 1.5-5-5v-.79l-.27-.27A6.516 6.516 0 0 1 9.5 16 6.5 6.5 0 0 1 3 9.5 6.5 6.5 0 0 1 9.5 3m0 2C7 5 5 7 5 9.5S7 14 9.5 14 14 12 14 9.5 12 5 9.5 5z"/></svg>
      </label>
      <div class="md-search" data-md-component="search" role="dialog">
  <label class="md-search__overlay" for="__search"></label>
  <div class="md-search__inner" role="search">
    <form class="md-search__form" name="search">
      <input type="text" class="md-search__input" name="query" aria-label="Search" placeholder="Search" autocapitalize="off" autocorrect="off" autocomplete="off" spellcheck="false" data-md-component="search-query" required>
      <label class="md-search__icon md-icon" for="__search">
        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M9.5 3A6.5 6.5 0 0 1 16 9.5c0 1.61-.59 3.09-1.56 4.23l.27.27h.79l5 5-1.5 1.5-5-5v-.79l-.27-.27A6.516 6.516 0 0 1 9.5 16 6.5 6.5 0 0 1 3 9.5 6.5 6.5 0 0 1 9.5 3m0 2C7 5 5 7 5 9.5S7 14 9.5 14 14 12 14 9.5 12 5 9.5 5z"/></svg>
        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M20 11v2H8l5.5 5.5-1.42 1.42L4.16 12l7.92-7.92L13.5 5.5 8 11h12z"/></svg>
      </label>
      <nav class="md-search__options" aria-label="Search">
        
        <button type="reset" class="md-search__icon md-icon" aria-label="Clear" tabindex="-1">
          <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M19 6.41 17.59 5 12 10.59 6.41 5 5 6.41 10.59 12 5 17.59 6.41 19 12 13.41 17.59 19 19 17.59 13.41 12 19 6.41z"/></svg>
        </button>
      </nav>
      
        <div class="md-search__suggest" data-md-component="search-suggest"></div>
      
    </form>
    <div class="md-search__output">
      <div class="md-search__scrollwrap" data-md-scrollfix>
        <div class="md-search-result" data-md-component="search-result">
          <div class="md-search-result__meta">
            Initializing search
          </div>
          <ol class="md-search-result__list"></ol>
        </div>
      </div>
    </div>
  </div>
</div>
    
    
  </nav>
  
</header>
    
    <div class="md-container" data-md-component="container">
      
      
        
          
            
<nav class="md-tabs" aria-label="Tabs" data-md-component="tabs">
  <div class="md-tabs__inner md-grid">
    <ul class="md-tabs__list">
      
        
  
  


  
  
  
    <li class="md-tabs__item">
      <a href="../../../.." class="md-tabs__link">
        Getting Started
      </a>
    </li>
  

      
    </ul>
  </div>
</nav>
          
        
      
      <main class="md-main" data-md-component="main">
        <div class="md-main__inner md-grid">
          
            
              
              <div class="md-sidebar md-sidebar--primary" data-md-component="sidebar" data-md-type="navigation" >
                <div class="md-sidebar__scrollwrap">
                  <div class="md-sidebar__inner">
                    

  


  

<nav class="md-nav md-nav--primary md-nav--lifted md-nav--integrated" aria-label="Navigation" data-md-level="0">
  <label class="md-nav__title" for="__drawer">
    <a href="../../../.." title="WL4G DOCS" class="md-nav__button md-logo" aria-label="WL4G DOCS" data-md-component="logo">
      
  <img src="../../../../static/images/mylogo.jpeg" alt="logo">

    </a>
    WL4G DOCS
  </label>
  
  <ul class="md-nav__list" data-md-scrollfix>
    
      
      
      

  
  
  
    
    <li class="md-nav__item md-nav__item--nested">
      
      
        <input class="md-nav__toggle md-toggle" data-md-toggle="__nav_1" data-md-state="indeterminate" type="checkbox" id="__nav_1" checked>
      
      
      
      
        <label class="md-nav__link" for="__nav_1">
          Getting Started
          <span class="md-nav__icon md-icon"></span>
        </label>
      
      <nav class="md-nav" aria-label="Getting Started" data-md-level="1">
        <label class="md-nav__title" for="__nav_1">
          <span class="md-nav__icon md-icon"></span>
          Getting Started
        </label>
        <ul class="md-nav__list" data-md-scrollfix>
          
            
              
  
  
  
    <li class="md-nav__item">
      <a href="../../../.." class="md-nav__link">
        Introduction
      </a>
    </li>
  

            
          
            
              
  
  
  
    <li class="md-nav__item">
      <a href="../../../../ABOUT_CN/" class="md-nav__link">
        About
      </a>
    </li>
  

            
          
        </ul>
      </nav>
    </li>
  

    
  </ul>
</nav>
                  </div>
                </div>
              </div>
            
            
          
          <div class="md-content" data-md-component="content">
            <article class="md-content__inner md-typeset">
              
                


<h1 id="kafka-offset">生产环境 kafka 消费中断如何 offset 重放消费<a class="headerlink" href="#kafka-offset" title="Permanent link">&para;</a></h1>
<h2 id="1">1. 背景<a class="headerlink" href="#1" title="Permanent link">&para;</a></h2>
<p>&nbsp;&nbsp;&nbsp;&nbsp;晚上收到告警消息说没实时数据了...  架构是: <code>kafka -&gt; spark streaming -&gt; hbase</code> 消费数据存到 hbase 做实时历史查询, 由于在外面无法及时登录服务器... 最后经排查发现是 yarn 调度卡死导致 spark streaming 任务假死, 处理过程省略xx字...,  待一切恢复之后, 发现历史曲线凹下去一节, 推测肯定是处理问题这段时间数据没消费到, 讲道理这种场景不应该管理好 offset了, 即使中断了也应该从此出继续消费才是呀... 于是检查代码发现... 然后一百种想TMD... 前任留的坑呀... 只创建完<code>val kafkaStream=KafkaUtils.createStream(ssc, zkKafka, groupId, topicsMap)</code>就完事了, 根本没理 offset...</p>
<h2 id="2">2. 分析<a class="headerlink" href="#2" title="Permanent link">&para;</a></h2>
<p>我们在设计基于 kafka 消费系统时通常会按照业务特点，对应消费策略也分几种情况:</p>
<ul>
<li>
<p>a. 如果只需要最新数据那么消费策略通常是设置<code>auto.offset.reset</code>=<code>latest</code> (默认值);</p>
</li>
<li>
<p>b. 如果要求每次启动消费从头开始则需设置<code>auto.offset.reset</code>=<code>earliest</code>;</p>
</li>
<li>
<p>c. 例如业务是 kafka -&gt; spark streaming -&gt; hbase 做历史数据存储, 那么自动或手动(需设置<code>enable.auto.commit</code>=<code>false</code>)管理好每次处理完 fetch 来的数据后的 offset 了, 通常客户端都有自动提交和手动提交两种策略, 后者更加灵活如自行存到 redis / mysql / hbase 等。</p>
</li>
<li>
<p>a. 仅使用<code>val kafkaStream=KafkaUtils.createStream(ssc, zkKafka, groupId, topicsMap)</code>是不行的，上面介绍了默认情况下 KafkaConsumer 会使用<code>auto.offset.reset</code>=<code>latest</code>策略，</p>
</li>
</ul>
<h2 id="3">3. 解决<a class="headerlink" href="#3" title="Permanent link">&para;</a></h2>
<ul>
<li>3.1 首先确保 Spark Streaming 消费 Kafka 的模式支持 offset (即不是 <code>latest</code> 模式)，关键代码部分如下，使用 kafka 直接消费模式比 zk 模式性能更高, 同时 offset 的管理也直接提交到 kafka brokers 里, 减少使用 Redis / MySQL 等外部存储依赖. <a href="https://gitee.com/wl4g-private/xcloud-dataworks/blob/master/xcloud-dataworks-stream/src/main/scala/com/wl4g/dataworks/spark/streaming/GenericKafka2HbaseStreaming.scala">完整代码请参考</a></li>
</ul>
<details style="cursor:pointer;">
<summary>GenericKafka2HbaseStreaming.scala</summary>
<pre><code>
   // Skip ......

    val Array(master, brokers, hbaseZk, 
         groupId, topicTables, delayMs, rowKeyFields) = args
    val topicTablesMap = topicTables.split(",")
        .map(x => x.split("->")).map(x => Tuple2(x(0), x(1))).toMap

    val sparkConf = new SparkConf()
        .setAppName(getClass.getSimpleName).setMaster(master)
   val ssc = new StreamingContext(sparkConf, 
        Milliseconds(delayMs.toLong)).checkpoint("checkpoint")

    val kafkaDStream = Kafkas.createDirectStream(ssc, 
        brokers, groupId, topicTablesMap)

    kafkaDStream.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        rdd.foreachPartition(records => {
          if (records != null && !records.isEmpty) {
            val hbConn = HbaseUtils2.createConnDefault(hbaseZk)
            val batchStore = new BatchStore(hbConn, 500)
            records.foreach(record => {
                 // Skip ......
            })
            batchStore.commitAll()
            hbConn.close()
          }
        })

        Kafkas.commitAsync(kafkaDStream, rdd)
      }
    })
    ssc.start()
    ssc.awaitTermination()
</code></pre>
</details>
<details style="cursor:pointer;">
<summary>Kafkas.scala</summary>
<pre><code>
object Kafkas extends Logging with Serializable {

  /**
   * Create kafka direct input stream with brokers and topics.
   */
  def createDirectStream(ssc: StreamingContext, 
      brokers: String, groupId: String, 
      topicTablesMap: Map[String, String]):
      InputDStream[ConsumerRecord[String, String]] = {
    Asserts.hasText(brokers, 
        "'brokers' must not be empty, please check parameters.")
    Asserts.hasText(groupId, "groupId")
    Asserts.notEmpty(topicTablesMap, 
        "'topicTablesMap' must not be empty, please check parameters.")

    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> 
      classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> 
      classOf[StringDeserializer].getName,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ->
      Boolean.valueOf(System.getProperty("kafka.enable.auto.commit", "false")),
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> 
      System.getProperty("kafka.auto.offset.reset", "earliest")) // latest

    // Create direct kafka stream with brokers and topics
    val topics = topicTablesMap.keys.toArray[String]
    Asserts.notNull(topics, "Topics must not be null")
    Asserts.notNullElements(topics, "Topics must elements not be null")

    KafkaUtils.createDirectStream(
      ssc, LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
  }

  // Skip ......

  /**
   * Commit offset RDD.</br>
   */
  def commitAsync(
    kafkaDStream: InputDStream[ConsumerRecord[String, String]],
    rdd: RDD[ConsumerRecord[String, String]]) {
    // Commit current offsetRanages.
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    kafkaDStream.asInstanceOf[CanCommitOffsets]
      .commitAsync(offsetRanges, new OffsetCommitCallback() {
          def onComplete(m: java.util.Map[TopicPartition, OffsetAndMetadata], ex: Exception) {
            if (ex != null) {
              log.error(String.format(
                "Failed to async commit kafka brokers. offsetRanges - {}", offsetRanges.toList), ex)
            } else if (log.isInfoEnabled()) {
              log.info("Committed offsetRanges - {}", offsetRanges.toList)
            }
          }
    })
  }

}
</code></pre>
</details>

<p></br></p>
<ul>
<li>
<p>3.2 使用 <a href="https://github.com/wl4g/kafka_offset_tool">kafka_offset_tool</a> 工具将 对应 group 的 topics 的 offset 重置到故障之前</p>
</li>
<li>
<p>3.2.1 下载工具，注意操作系统版本，其他操作系统需自行编译</p>
</li>
</ul>
<div class="highlight"><pre><span></span><code><a id="__codelineno-0-1" name="__codelineno-0-1"></a><a href="#__codelineno-0-1"><span class="linenos" data-linenos="1 "></span></a>wget https://github.com/wl4g/kafka_offset_tool/releases/download/v1.2.5/kafkaOffsetTool_1.2.5_linux_amd64
<a id="__codelineno-0-2" name="__codelineno-0-2"></a><a href="#__codelineno-0-2"><span class="linenos" data-linenos="2 "></span></a>chmod +x kafkaOffsetTool_1.2.5_linux_amd64
</code></pre></div>
<ul>
<li>3.2.2 查看当前 groups、topics、offsets 信息，支持多种过滤，具体用法参见:
 https://github.com/wl4g/kafka_offset_tool</li>
</ul>
<div class="highlight"><pre><span></span><code><a id="__codelineno-1-1" name="__codelineno-1-1"></a><a href="#__codelineno-1-1"><span class="linenos" data-linenos="1 "></span></a>./kafkaOffsetTool_1.2.5_linux_amd64 get-offset -o myoffset.json
</code></pre></div>
<ul>
<li>
<p>3.2.3 (可选) 可手动编辑 <code>myoffset.json</code>，对 <code>group / topic / offset</code> 做特殊删除或修改</p>
</li>
<li>
<p>3.3.4 使用 <code>offset-calc</code> 命令对 json 文件批量重置 offset 。如下示例是将所有groupId 对所有 topics 的 offset 的值全部回放 <code>1000</code>，即都减 1000，具体多少可根据业务量估算，也可通过 <code>kafka-console-consumer.sh</code> 从头消费查查看，一般存到主题的消息都带有时间，即根据它可以知道目前还没有被 Kafka 清除的最早的数据是什么时候的。</p>
</li>
</ul>
<div class="highlight"><pre><span></span><code><a id="__codelineno-2-1" name="__codelineno-2-1"></a><a href="#__codelineno-2-1"><span class="linenos" data-linenos="1 "></span></a>./kafkaOffsetTool_1.2.5_linux_amd64 offset-calc -i myoffset.json -o myoffset-restore.json -I -1000
</code></pre></div>
<ul>
<li>3.2.5 【重点】重置 offset 到指定值</li>
</ul>
<div class="highlight"><pre><span></span><code><a id="__codelineno-3-1" name="__codelineno-3-1"></a><a href="#__codelineno-3-1"><span class="linenos" data-linenos="1 "></span></a>./kafkaOffsetTool_1.2.5_linux_amd64 set-offset -i myoffset-restore.json
</code></pre></div>
<ul>
<li>3.2.6 再重启 Spark Streaming 消费程序，观察消费的数据是否从重置的 offset 的时间开始的</li>
</ul>
<h2 id="4">4. 参考资料<a class="headerlink" href="#4" title="Permanent link">&para;</a></h2>
<p><a href="https://blog.csdn.net/xiaohu21/article/details/110757083">参考1: spark streaming 结合kafka 精确消费一次将结果保存到hbase</a></p>

              
            </article>
          </div>
        </div>
        
      </main>
      
        <footer class="md-footer">
  
  <div class="md-footer-meta md-typeset">
    <div class="md-footer-meta__inner md-grid">
      <div class="md-copyright">
  
  
    Made with
    <a href="https://squidfunk.github.io/mkdocs-material/" target="_blank" rel="noopener">
      Material for MkDocs
    </a>
  
</div>
      
    </div>
  </div>
</footer>
      
    </div>
    <div class="md-dialog" data-md-component="dialog">
      <div class="md-dialog__inner md-typeset"></div>
    </div>
    <script id="__config" type="application/json">{"base": "../../../..", "features": ["search.suggest", "search.highlight", "navigation.tabs", "navigation.expand", "toc.follow", "toc.integrate"], "search": "../../../../assets/javascripts/workers/search.5e67fbfe.min.js", "translations": {"clipboard.copied": "Copied to clipboard", "clipboard.copy": "Copy to clipboard", "search.config.lang": "en", "search.config.pipeline": "trimmer, stopWordFilter", "search.config.separator": "[\\s\\-]+", "search.placeholder": "Search", "search.result.more.one": "1 more on this page", "search.result.more.other": "# more on this page", "search.result.none": "No matching documents", "search.result.one": "1 matching document", "search.result.other": "# matching documents", "search.result.placeholder": "Type to start searching", "search.result.term.missing": "Missing", "select.version.title": "Select version"}, "version": {"default": "latest", "provider": "mike"}}</script>
    
    
      <script src="../../../../assets/javascripts/bundle.c44cc438.min.js"></script>
      
        <script src="../../../../static/js/util.js"></script>
      
    
  </body>
</html>